package bigdata.mapreduce.relation;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class Reduce extends Reducer<Text,Text,Text,NullWritable> {
	private Text key=new Text();
	NullWritable nullvalue;

	@Override
	public void reduce(Text key,Iterable<Text> values, Context context) throws InterruptedException, IOException {
		List<String> R=new ArrayList<>();
		List<String> S=new ArrayList<>();
		
		for(Text val:values) {
			System.out.println(key.toString()+"\t"+val.toString());
			String name=val.toString().split(",")[0];
			
			if(val.toString().contains(",R"))
				R.add(name);
			else if(val.toString().contains(",S"))
				S.add(name);
		}
		
		for(String sr:R) {
			for(String ss:S) {
				this.key.set(sr + "<==>" + ss);
				context.write(this.key, nullvalue);
			}
		}
		
	}

}
